package com.cl.util

import java.sql.{Connection, DriverManager, PreparedStatement}

import com.cl.bean.WaterQualityBean
import org.apache.spark.sql.{ForeachWriter, Row}
import org.apache.spark.internal.Logging

/**
 * @Author xc
 * @Date 2023/4/17 16:52
 * @Version 1.0
 */
//针对sd_d_p
abstract class  BaseJdbcSink (sql: String) extends ForeachWriter[WaterQualityBean] {
  def realProcess(str: String,  bean: WaterQualityBean)

  var conn: Connection = _
  var ps: PreparedStatement = _
  val driver = "com.mysql.cj.jdbc.Driver"

  //开启连接
  override def open(partitionId: Long, version: Long): Boolean = {
    print("open....")
    Class.forName(driver)
  conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/ym?characterEncoding=UTF-8", "root", "cl416416")
    print("open....xxx")
  true
}

  //处理数据--将数据存入到MySQL
  override def process(bean: WaterQualityBean): Unit = {
    try {
      println("process....")
      realProcess(sql, bean)
      println("process....xxxx")
    } catch {
      case e: Exception =>
        e.printStackTrace()
    }
//    print("process....")
//  realProcess(sql,bean )
//    print("process....xxxx")
}

  //关闭连接
  override def close(errorOrNull: Throwable): Unit = {
    print("close....xxxx")
  if (conn != null) {
  conn.close()
}
  if (ps != null) {
  ps.close()
}
}
}
